/**
 * Copyright 2016 vip.com.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
 * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations under the License.
 * </p>
 **/

package spring.cloud.tasks.api.service.impl.statistics.analyzer;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;
import spring.cloud.tasks.api.domain.*;
import spring.cloud.tasks.api.service.TaskService;
import spring.cloud.tasks.api.utils.TaskNodePath;
import spring.cloud.tasks.api.zookeeper.ZooKeeper;

import java.util.ArrayList;
import java.util.List;

/**
 * @author timmy.hu
 */
public class UnableFailoverJobAnalyzer {

    private static final Logger log = LoggerFactory.getLogger(UnableFailoverJobAnalyzer.class);
    private final List<AbnormalJob> unableFailoverJobList = new ArrayList<AbnormalJob>();
    private TaskService executorProvidedList;

    /**
     * 查找无法高可用的作业
     */
    public void analyze(ZooKeeper zooKeeper, String jobName, String jobDegree,
                        NameSpaceConfiguration config) {
        AbnormalJob unableFailoverJob = new AbnormalJob(jobName, config.getNameSpace(),
                config.getDegree());
        if (isUnableFailoverJob(zooKeeper, unableFailoverJob)) {
            unableFailoverJob.setJobDegree(jobDegree);
            addUnableFailoverJob(unableFailoverJob);
        }
    }

    private synchronized void addUnableFailoverJob(AbnormalJob unableFailoverJob) {
        unableFailoverJobList.add(unableFailoverJob);
    }

    /**
     * 无法高可用的情况： 1、勾选只使用优先executor，preferList只有一个物理机器（剔除offline、deleted的物理机） 2、没有勾选只使用优先executor，没有选择容器资源，可供选择的preferList只有一个物理机器（剔除offline、deleted的物理机，剔除容器资源）
     */
    private boolean isUnableFailoverJob(ZooKeeper zooKeeper, AbnormalJob unableFailoverJob) {
        try {
            String jobName = unableFailoverJob.getJobName();
            List<ExecutorProvided> preferListProvided = executorProvidedList.getExecutorProvidedList(zooKeeper.getCuratorFramework().getNamespace(), jobName);
            if (CollectionUtils.isEmpty(preferListProvided)) {
                return false;
            }
            String preferList = zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "preferList"));
            List<String> preferListArr = toPerferListArr(preferList);
            boolean containerSelected = false;
            int count = 0;
            if (onlyUsePreferList(zooKeeper, jobName)) {
                for (ExecutorProvided executorProvided : preferListProvided) {
                    if (!preferListArr.contains(executorProvided.getExecutorId())) {
                        continue;
                    }
                    if (ExecutorType.CONTAINER.equals(executorProvided.getExecutorType())) {
                        containerSelected = true;
                        break;
                    } else if (ExecutorType.PHYSICAL_MACHINE.equals(executorProvided.getExecutorType())
                            && ExecutorProvidedStatus.ONLINE.equals(executorProvided.getExecutorProvidedStatus())) {
                        count++;
                    }
                }
            } else {
                for (ExecutorProvided executorProvided : preferListProvided) {
                    if (preferListArr.contains(executorProvided.getExecutorId())
                            && ExecutorType.CONTAINER.equals(executorProvided.getExecutorType())) {
                        containerSelected = true;
                        break;
                    }
                    if (ExecutorType.PHYSICAL_MACHINE.equals(executorProvided.getExecutorType()) &&
                            ExecutorProvidedStatus.ONLINE.equals(executorProvided.getExecutorProvidedStatus())) {
                        count++;
                    }
                    if (count > 1) {
                        break;
                    }
                }
            }
            return !containerSelected && count == 1;
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            return false;
        }
    }

    private boolean onlyUsePreferList(ZooKeeper zooKeeper, String jobName) {
        String useDispreferListStr = zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "useDispreferList"));
        return useDispreferListStr != null && !Boolean.parseBoolean(useDispreferListStr);
    }

    private List<String> toPerferListArr(String preferList) {
        List<String> preferListArr = new ArrayList<>();
        if (StringUtils.isBlank(preferList)) {
            return preferListArr;
        }
        String[] split = preferList.split(",");
        for (String prefer : split) {
            String tmp = prefer.trim();
            if (tmp.length() > 0 && !preferListArr.contains(tmp)) {
                preferListArr.add(tmp);
            }
        }
        return preferListArr;
    }

    public List<AbnormalJob> getUnableFailoverJobList() {
        return new ArrayList<AbnormalJob>(unableFailoverJobList);
    }

    public void setExecutorProvidedList(TaskService executorProvidedList) {
        this.executorProvidedList = executorProvidedList;
    }
}
